package com.idanchuang.count.core.deserializer;

import com.idanchuang.component.core.util.StringUtil;
import com.idanchuang.count.core.entity.SimpleCountIncrEvent;
import com.idanchuang.count.core.util.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

/**
 * @author Richard_yyf
 * @version 1.0 2021/9/24
 */
@Slf4j
public class SimpleCountIncrEventDeserializer implements Deserializer<SimpleCountIncrEvent> {


    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public SimpleCountIncrEvent deserialize(String topic, byte[] data) {

        SimpleCountIncrEvent event;
        try {
            event = JacksonUtil.parseObject(data, SimpleCountIncrEvent.class);
        } catch (Exception e) {
            log.error("SimpleCountIncrEvent 反序列化失败", e);
            return null;
        }

        if (StringUtil.isBlank(event.getBizNo()) ||
                StringUtil.isBlank(event.getBizCode()) || event.getIncrement() == null) {
            log.error("接收到的 SimpleCountIncrEvent 必填字段为空：{}", new String(data));
            return null;
        }

        return event;
    }

    @Override
    public void close() {

    }
}
